"""
语音服务模块

提供语音转文本和文本转语音功能
"""

import os
import logging
import tempfile
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List, Tuple

# 强制PyTorch使用CPU
os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1'
os.environ['PYTORCH_MPS_HIGH_WATERMARK_RATIO'] = '0.0'

from src.core.config_manager import ConfigManager

logger = logging.getLogger(__name__)

class SpeechToText(ABC):
    """语音转文本接口"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        self.config = config
        self.debug = debug
        self.logger = logger
        
        if self.debug:
            self.logger.setLevel(logging.DEBUG)
    
    @abstractmethod
    def transcribe_audio(self, audio_file_path: str) -> str:
        """
        将语音文件转换为文本
        
        Args:
            audio_file_path: 音频文件路径
            
        Returns:
            转换后的文本
        """
        pass

class TextToSpeech(ABC):
    """文本转语音接口"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        self.config = config
        self.debug = debug
        self.logger = logger
        
        if self.debug:
            self.logger.setLevel(logging.DEBUG)
    
    @abstractmethod
    def synthesize_speech(self, text: str, output_file: Optional[str] = None) -> str:
        """
        将文本转换为语音
        
        Args:
            text: 要转换的文本
            output_file: 输出文件路径，如果为None则生成临时文件
            
        Returns:
            生成的音频文件路径
        """
        pass

class WhisperSTT(SpeechToText):
    """使用OpenAI Whisper模型的语音转文本实现"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        super().__init__(config, debug)
        self.model = config.get("model", "base")
        self.language = config.get("language", "zh")
        self.logger.debug(f"初始化Whisper STT: 模型={self.model}, 语言={self.language}")
        
        try:
            import whisper
            self.whisper_model = whisper.load_model(self.model)
        except ImportError:
            self.logger.error("请安装OpenAI Whisper: pip install openai-whisper")
            raise
    
    def transcribe_audio(self, audio_file_path: str) -> str:
        """
        将语音文件转换为文本
        
        Args:
            audio_file_path: 音频文件路径
            
        Returns:
            转换后的文本
        """
        if not os.path.exists(audio_file_path):
            self.logger.error(f"音频文件不存在: {audio_file_path}")
            return "错误：音频文件不存在"
        
        try:
            self.logger.debug(f"开始转写音频: {audio_file_path}")
            result = self.whisper_model.transcribe(
                audio_file_path,
                language=self.language
            )
            
            transcription = result["text"].strip()
            self.logger.debug(f"转写完成: {transcription[:50]}...")
            return transcription
            
        except Exception as e:
            self.logger.error(f"音频转写失败: {e}")
            return f"音频转写失败：{str(e)}"

class EdgeTTS(TextToSpeech):
    """使用Microsoft Edge TTS的文本转语音实现"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        super().__init__(config, debug)
        self.voice = config.get("voice", "zh-CN-XiaoxiaoNeural")
        self.rate = config.get("rate", "+0%")
        self.volume = config.get("volume", "+0%")
        self.logger.debug(f"初始化Edge TTS: 语音={self.voice}, 语速={self.rate}")
        
        try:
            import edge_tts
            self.communicate_class = edge_tts.Communicate
        except ImportError:
            self.logger.error("请安装edge-tts: pip install edge-tts")
            raise
    
    async def _synthesize_async(self, text: str, output_file: str) -> None:
        """异步生成语音"""
        communicate = self.communicate_class(
            text, 
            self.voice,
            rate=self.rate,
            volume=self.volume
        )
        
        try:
            await communicate.save(output_file)
        except Exception as e:
            self.logger.error(f"语音合成异步操作失败: {e}")
            raise
    
    def synthesize_speech(self, text: str, output_file: Optional[str] = None) -> str:
        """
        将文本转换为语音
        
        Args:
            text: 要转换的文本
            output_file: 输出文件路径，如果为None则生成临时文件
            
        Returns:
            生成的音频文件路径
        """
        import asyncio
        
        if not text:
            self.logger.warning("收到空文本，无法生成语音")
            return ""
        
        if output_file is None:
            # 创建临时文件
            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
            output_file = temp_file.name
            temp_file.close()
            
        self.logger.debug(f"开始合成语音: 文本长度={len(text)}, 输出文件={output_file}")
        
        try:
            # 运行异步函数
            asyncio.run(self._synthesize_async(text, output_file))
            self.logger.debug(f"语音合成完成: {output_file}")
            return output_file
        except Exception as e:
            self.logger.error(f"语音合成失败: {e}")
            if output_file and os.path.exists(output_file):
                try:
                    os.remove(output_file)
                except:
                    pass
            return ""

class MockSTT(SpeechToText):
    """模拟的语音转文本服务，用于测试"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        super().__init__(config, debug)
        self.logger.debug("初始化模拟STT服务")
    
    def transcribe_audio(self, audio_file_path: str) -> str:
        """
        模拟将语音文件转换为文本
        
        Args:
            audio_file_path: 音频文件路径
            
        Returns:
            固定的示例文本
        """
        self.logger.debug(f"模拟转写音频: {audio_file_path}")
        return "这是一个模拟的语音转文本结果。实际的音频文件是: " + os.path.basename(audio_file_path)

class MockTTS(TextToSpeech):
    """模拟的文本转语音服务，用于测试"""
    
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        super().__init__(config, debug)
        self.logger.debug("初始化模拟TTS服务")
    
    def synthesize_speech(self, text: str, output_file: Optional[str] = None) -> str:
        """
        模拟将文本转换为语音
        
        Args:
            text: 要转换的文本
            output_file: 输出文件路径，如果为None则生成临时文件
            
        Returns:
            生成的音频文件路径（或表示成功的消息）
        """
        if output_file is None:
            # 创建临时文件
            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
            output_file = temp_file.name
            temp_file.close()
            
        self.logger.debug(f"模拟生成语音: 文本={text[:30]}..., 输出={output_file}")
        
        try:
            # 生成一个简单的无声WAV文件，而不是创建一个空文件
            import wave
            import struct
            
            # 参数设置
            duration = 2  # 秒
            frequency = 440  # Hz
            sample_rate = 44100  # 采样率
            
            # 创建无声WAV文件
            with wave.open(output_file, 'w') as wav_file:
                wav_file.setparams((1, 2, sample_rate, 0, 'NONE', 'not compressed'))
                
                # 创建一个全零的数据块，表示无声
                for i in range(0, int(duration * sample_rate)):
                    value = 0  # 无声
                    packed_value = struct.pack('h', value)
                    wav_file.writeframes(packed_value)
            
            return output_file
        except Exception as e:
            self.logger.error(f"生成模拟音频文件失败: {e}")
            return ""

class MeloTTS(TextToSpeech):
    """使用 melo 库的文本转语音实现"""
    def __init__(self, config: Dict[str, Any], debug: bool = False):
        super().__init__(config, debug)
        self.language = config.get("language", "ZH")
        self.speaker = config.get("speaker", None)  # 可选：指定说话人
        self.logger.debug(f"初始化Melo TTS: language={self.language}, speaker={self.speaker}")
        try:
            # 完全禁用MPS设备
            import torch
            if torch.backends.mps.is_available():
                torch.backends.mps.enabled = False
            torch.set_default_device('cpu')
            torch.set_default_tensor_type(torch.FloatTensor)
            
            try:
                from melo.api import TTS
            except ImportError:
                try:
                    from melo.tts import TTS
                except ImportError:
                    raise ImportError("你的 melo 安装不包含 TTS 类，请升级 melo 或参考官方文档。")
            self.melo_tts_class = TTS
        except ImportError as e:
            self.logger.error(f"请先安装 melo 并确保包含 TTS 类: {e}")
            raise

    def synthesize_speech(self, text: str, output_file: Optional[str] = None) -> str:
        """
        使用 melo 生成语音文件
        Args:
            text: 要转换的文本
            output_file: 输出文件路径，如果为None则生成临时文件
        Returns:
            生成的音频文件路径
        """
        if not text:
            self.logger.warning("收到空文本，无法生成语音")
            return ""
        if output_file is None:
            import tempfile
            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
            output_file = temp_file.name
            temp_file.close()
        try:
            # 确保MPS设备被禁用
            import torch
            if torch.backends.mps.is_available():
                torch.backends.mps.enabled = False
            torch.set_default_device('cpu')
            torch.set_default_tensor_type(torch.FloatTensor)
            
            tts = self.melo_tts_class(language=self.language, device='cpu')
            # 获取默认说话人ID
            speaker_id = 0
            if self.speaker and self.speaker in tts.hps.data.spk2id:
                speaker_id = tts.hps.data.spk2id[self.speaker]
            else:
                # 如果没有指定说话人或说话人不存在，使用第一个可用的说话人
                speaker_id = list(tts.hps.data.spk2id.values())[0]
            
            # 使用正确的参数调用tts_to_file
            tts.tts_to_file(text=text, speaker_id=speaker_id, output_path=output_file)
            self.logger.debug(f"Melo TTS 语音合成完成: {output_file}")
            return output_file
        except Exception as e:
            self.logger.error(f"Melo TTS 语音合成失败: {e}")
            return ""

class SpeechServiceFactory:
    """语音服务工厂，用于创建STT和TTS服务实例"""
    
    @staticmethod
    def create_stt(config: Dict[str, Any], debug: bool = False) -> SpeechToText:
        """
        创建语音转文本服务实例
        
        Args:
            config: 配置字典
            debug: 是否开启调试模式
            
        Returns:
            SpeechToText实例
            
        Raises:
            ValueError: 如果STT类型不支持
        """
        stt_type = config.get("type", "mock").lower()
        
        if stt_type == "whisper":
            return WhisperSTT(config, debug)
        elif stt_type == "mock":
            return MockSTT(config, debug)
        else:
            raise ValueError(f"不支持的STT类型: {stt_type}")
    
    @staticmethod
    def create_tts(config: Dict[str, Any], debug: bool = False) -> TextToSpeech:
        """
        创建文本转语音服务实例
        
        Args:
            config: 配置字典
            debug: 是否开启调试模式
            
        Returns:
            TextToSpeech实例
            
        Raises:
            ValueError: 如果TTS类型不支持
        """
        tts_type = config.get("type", "mock").lower()
        
        if tts_type == "edge_tts":
            return EdgeTTS(config, debug)
        elif tts_type == "melo":
            return MeloTTS(config, debug)
        elif tts_type == "mock":
            return MockTTS(config, debug)
        else:
            raise ValueError(f"不支持的TTS类型: {tts_type}")
